package com.atguigu.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.util.DateFormatUtil;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

public class DwdTrafficUvApp {
    public static void main(String[] args) throws Exception {

        //1  环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        String topic="dwd_traffic_page_log";
        String groupId="dwd_traffic_uv";

        // 2  数据来源： dwd_traffic_page_log 主题
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        DataStreamSource<String> kafkaStream = env.addSource(kafkaConsumer);

        //3  把数据转为jsonobj
        SingleOutputStreamOperator<JSONObject> jsonObjStream = kafkaStream.map(jsonString -> JSON.parseObject(jsonString));


        KeyedStream<JSONObject, String> midKeyedStream = jsonObjStream.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));

        //5   如何从页面日志中筛选出当日首次访问
        //      先筛选出本次会话的首页 last_page_id= null
        //       利用历史状态保存同一mid的【最后访问日期】
        //            如果从状态中取得的【最后访问日期】与当前日志的时间戳相同 ，认为该mid已经访问过了 ，过滤掉该数据。
        SingleOutputStreamOperator<JSONObject> uvFilteredStream = midKeyedStream.filter(new RichFilterFunction<JSONObject>() {

            ValueStateDescriptor<String> lastVisitDateStateDesc = new ValueStateDescriptor("last_visit_date", String.class);

            ValueState<String> lastVisitDateState = null;


            @Override
            public void open(Configuration parameters) throws Exception {
                //设定状态的声明周期  // 状态后端选什么？ 没有历史数据，生命周期短 可以考虑用hashmapStatebackend
                lastVisitDateStateDesc.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                lastVisitDateState = getRuntimeContext().getState(lastVisitDateStateDesc);
            }

            @Override
            public boolean filter(JSONObject jsonObject) throws Exception {

                //       利用历史状态保存同一mid的【最后访问日期】
                //            如果从状态中取得的【最后访问日期】与当前日志的时间戳相同 ，认为该mid已经访问过了 ，过滤掉该数据。
                String lastPageId = jsonObject.getJSONObject("page").getString("last_page_id");
                if (lastPageId != null && lastPageId.length() > 0) { // 先筛选出本次会话的首页 last_page_id= null
                    return false;
                }

                Long ts = jsonObject.getLong("ts");
                String curVisitDate = DateFormatUtil.toDate(ts);

                String lastVisitDate = lastVisitDateState.value();
                if (lastVisitDate == null || lastVisitDate.length() == 0) {
                    //  a 如果不存在访问时间   该访客未访问过  保留该用户 ，同时要把当前访问日期更新到状态中
                    lastVisitDateState.update(curVisitDate);
                    return true;
                } else {
                    //  b 如果存在访问时间
                    //
                    if (lastVisitDate.equals(curVisitDate)) {
                        //b1  如果和当前访问日期相同  认为今天已经访问过了 ，过滤掉该次访问
                        return false;
                    } else {
                        // b2  如果状态中最后访问日期 早于 当前访问日期
                        //    认为是之前访问过 今天未访问所以保留 ，同时更新当前状态。
                        lastVisitDateState.update(curVisitDate);
                        return true;
                    }
                }

            }
        });


        //6    然后直接写到 dwd_traffic_uv 主题即可。
        uvFilteredStream.print();
        uvFilteredStream.map(jsonObj->jsonObj.toJSONString()).addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_uv"));

        env.execute();

    }
}
